package baiduschPro;

import java.io.InputStream;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.apache.nifi.annotation.behavior.ReadsAttribute;
import org.apache.nifi.annotation.behavior.ReadsAttributes;
import org.apache.nifi.annotation.behavior.WritesAttribute;
import org.apache.nifi.annotation.behavior.WritesAttributes;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.SeeAlso;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.PropertyValue;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessorInitializationContext;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.stream.io.ByteArrayOutputStream;

import net.minidev.json.JSONArray;
import net.minidev.json.JSONObject;
import net.minidev.json.JSONValue;


@Tags({"download"})
@CapabilityDescription("donwload one page  ,input content,可多条，可单挑，例如： {\"url\":\"http://ddfdf.com/inn\",\"referurl\":\"http://dfjdhfjdf\"}")
@SeeAlso({})
@ReadsAttributes({@ReadsAttribute(attribute="", description="")})
@WritesAttributes({@WritesAttribute(attribute="", description="")})
public class DownLoadBatchGETProcessor extends AbstractProcessor {

	public static final PropertyDescriptor UserAgent = new PropertyDescriptor
            .Builder().name("UserAgent")
            .displayName("useragent")
            .description("useragent")
            .required(true)
            .defaultValue("Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/51.0.2704.84 Safari/537.36")
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();
	public static final PropertyDescriptor DownInterval = new PropertyDescriptor
            .Builder().name("DownInterval")
            .displayName("down interval")
            .description("down time interval,second")
            .required(true)
            .defaultValue("3")
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();
	
	
	 public static final Relationship MY_RELATIONSHIPOK = new Relationship.Builder()
	            .name("gen success")
	            .description("success relationship")
	            .build();
	    
	    public static final Relationship MY_RELATIONSHIPNO = new Relationship.Builder()
	            .name("gen failure")
	            .description("failure relationship")
	            .build();

	    private List<PropertyDescriptor> descriptors;

	    private Set<Relationship> relationships;

	    @Override
	    protected void init(final ProcessorInitializationContext context) {
	        final List<PropertyDescriptor> descriptors = new ArrayList<PropertyDescriptor>();
	        descriptors.add(DownInterval);
	        descriptors.add(UserAgent);
	        this.descriptors = Collections.unmodifiableList(descriptors);

	        final Set<Relationship> relationships = new HashSet<Relationship>();
	        relationships.add(MY_RELATIONSHIPOK);
	        relationships.add(MY_RELATIONSHIPNO);
	        this.relationships = Collections.unmodifiableSet(relationships);
	    }
	    
	    @Override
	    public Set<Relationship> getRelationships() {
	        return this.relationships;
	    }

	    @Override
	    public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
	        return descriptors;
	    }

	    @OnScheduled
	    public void onScheduled(final ProcessContext context) {

	  //  	System.out.println("onScheduled is clled");
	    }
	    
	public DownLoadBatchGETProcessor() {
		// TODO Auto-generated constructor stub
	}

	@Override
	public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
		// TODO Auto-generated method stub
			FlowFile flowFile = session.get();
	        if ( flowFile == null ) {
	        	
	        	System.out.println("flowfile is null");
	           return;
	        }        
			//读取FlowFile
	        String content = "";
	        InputStream in = session.read(flowFile);
	        try {
				byte[] by = readStream(in);
				content = new String(by,"UTF-8");
				
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
	        
	      //获取配置属性值
	        PropertyValue useragentp = context.getProperty("UserAgent");
	        String useragent = useragentp.getValue();
	        
	        PropertyValue timeintervalp = context.getProperty("DownInterval");
	        String timeinterval = timeintervalp.getValue();
	        if(timeinterval == null)
	        	timeinterval = "3";
	        if(timeinterval.isEmpty())
	        	timeinterval = "3";
	        int interval = Integer.parseInt(timeinterval);
	        
	        //输入的内容可能是多条也可能是一条
	        
	        JSONArray urls = getUrls(content);
	        int nsize = urls.size();
	        for(int i = 0; i < nsize; i++)
	        {
	        	JSONObject obj = (JSONObject) urls.get(i);
	        	String url = (String) obj.get("url");
	        	String referurl = (String)obj.get("referurl");
	        	
	        	//下载
	        	
	        	
	        	if(i == 0)
	        	{
	        		
	        	}
	        	else
	        	{
	        		
	        	}
	        }//for i
	        
	        
			///
		}
		
		public JSONArray getUrls(String strContent)
		{
			
			JSONArray objret = null;
			Object obj = JSONValue.parse(strContent);
			String strclassname = obj.getClass().getName();
			if(strclassname.equalsIgnoreCase("net.minidev.json.JSONArray"))
			{
				objret = (JSONArray)obj;
			}
			else if(strclassname.equalsIgnoreCase("net.minidev.json.JSONObject"))
			{
				objret = new JSONArray();
				objret.add(obj);
			}
			else
			{
				objret = new JSONArray();
			}
			return objret;
			
		}
	
		public static byte[] readStream(InputStream inStream) throws Exception {  
		        ByteArrayOutputStream outSteam = new ByteArrayOutputStream();  
		        byte[] buffer = new byte[1024];  
		        int len = -1;  
		        while ((len = inStream.read(buffer)) != -1) {  
		            outSteam.write(buffer, 0, len);  
		        }  
		        outSteam.close();  
		        inStream.close();  
		        return outSteam.toByteArray();  
		    }  


}
